#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <limits.h>
#include <dirent.h>
#include <sys/time.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/socket.h>

#include "data_type.h"
#include "alloc.h"
#include "memfunc.h"
#include "json.h"
#include "struct_deal.h"
#include "crypto_func.h"
#include "basefunc.h"
#include "memdb.h"
#include "message.h"
#include "connector.h"
#include "ex_module.h"
#include "channel.h"
#include "sys_func.h"

#include "record2msg.h"

#define MAX_LINE_LEN 1024

static unsigned char Buf[DIGEST_SIZE*128];
static BYTE * ReadBuf=Buf;
static int readbuf_len=0;
static BYTE * WriteBuf=Buf+DIGEST_SIZE*64;
static int write_len=0;

static int index = 0;
static char errorbuf[1024];
static unsigned char sendbuf[4096];
static CHANNEL * record2msg;
static void * extend_template;

RECORD(MESSAGE,TYPES) in_type;
RECORD(MESSAGE,TYPES) out_type;
void * in_template;
void * out_template;

int _get_types_from_string( RECORD(MESSAGE,TYPES) * types,char * string)
{
   int ret;  
   int offset=0;
   char Buf[DIGEST_SIZE];

   if(string==NULL)
        return -EINVAL;
    
   offset=Getfiledfromstr(Buf,string,'(',DIGEST_SIZE);
   if(offset<0)
        return -EINVAL;
   offset++;
   ret=Getfiledfromstr(Buf,string+offset,',',DIGEST_SIZE);
   if(ret<0)
        return -EINVAL;
   offset+=ret+1;
  
   ret=memdb_get_typeno(Buf);
   if(ret<0)
        return -EINVAL;
   types->type=ret;

   ret=Getfiledfromstr(Buf,string+offset,')',DIGEST_SIZE);
   if(ret<0)
        return -EINVAL;
   offset+=ret;

   ret=memdb_get_subtypeno(types->type,Buf);
   if(ret<0)
        return -EINVAL;
   types->subtype=ret;
 
   return 0; 
}

int record2msg_init(void * sub_proc,void * para)
{
    int ret;
    struct record2msg_init_para * init_para=para;
    if(para==NULL)
	    return -EINVAL;
    record2msg=channel_find(init_para->channel_name);

    if(record2msg==NULL)
	    return -EINVAL;	

   ret=_get_types_from_string(&in_type,init_para->in_type);
   if(ret<0)
        return -EINVAL;

   ret=_get_types_from_string(&out_type,init_para->out_type);
   if(ret<0)
        return -EINVAL;
   
    in_template = memdb_get_template(in_type.type,in_type.subtype);   
    if(in_template==NULL)
        return -EINVAL;
    out_template = memdb_get_template(out_type.type,out_type.subtype);
    if(out_template==NULL)
        return -EINVAL;
        
    return 0;
}

int record2msg_start(void * sub_proc,void * para)
{
    int ret = 0, len = 0, i = 0, j = 0;
    int rc = 0;

    int offset=0;	
    int readbuf_len=0;
    int record_len=0;
	BYTE local_uuid[DIGEST_SIZE];	
    ret=proc_share_data_getvalue("uuid",local_uuid);
	if(ret<0)
		return ret;
    void * record_struct;
    void * send_msg;
    void * recv_msg;

    while(1)
    {
        usleep(time_val.tv_usec*10);
	    ret=channel_read(record2msg,ReadBuf+readbuf_len,DIGEST_SIZE*32-readbuf_len);
	    if(ret<0)
		    return ret;
        offset=0;
	    if(ret>0)
	    {
		    print_cubeaudit("record2msg read %d  %d data %s",ret,offset,ReadBuf+offset);
            readbuf_len+=ret;
            while(offset<readbuf_len)
            {
                record_struct = Talloc0(struct_size(out_template));
                if(record_struct==NULL)
                    return -ENOMEM;
                ret=blob_2_struct(ReadBuf+offset,record_struct,out_template);
                if(ret<0)
                {
                    if(readbuf_len-offset > 1440)
                        return -EINVAL;
                }
			    offset+=ret;
                send_msg=message_create(out_type.type,out_type.subtype,NULL);
                message_add_record(send_msg,record_struct);
                ex_module_sendmsg(sub_proc,send_msg);
            }
            if(offset>0)
            {
                Memcpy(ReadBuf,ReadBuf+offset,readbuf_len-offset);
                readbuf_len-=offset;
            }
		}
					
     	if((ex_module_recvmsg(sub_proc,&recv_msg)>=0)
		&&(recv_msg!=NULL))
     	{
            i=0;
            offset=0;
            while(message_get_record(recv_msg,&record_struct,i++)>=0)
            {
                if(record_struct==NULL)
                    break;
                ret=struct_2_blob(record_struct,WriteBuf+offset,in_template);
                if(ret<0)
                    return -EINVAL;
                offset+=ret;
            } 
	   	 	len=channel_write(record2msg,WriteBuf,offset);
            if (len != offset)
                print_cubeerr("record2msg write failed!\n");
			else
				print_cubeaudit("record2msg write %d data",offset);
        }
    }
    return 0;
}
